package com.sc.es.search;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.primitives.Ints;
import com.google.common.primitives.Longs;
import com.sc.es.common.*;
import com.sc.es.dto.HouseBucketDTO;
import com.sc.es.entity.House;
import com.sc.es.entity.HouseDetail;
import com.sc.es.entity.HouseTag;
import com.sc.es.entity.SupportAddress;
import com.sc.es.form.MapSearch;
import com.sc.es.repo.HouseDetailRepo;
import com.sc.es.repo.HouseRepo;
import com.sc.es.repo.HouseTagRepo;
import com.sc.es.repo.SupportAddressRepository;
import com.sc.es.service.IAddressService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.modelmapper.ModelMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * @author sc
 * Created on  2018/2/9
 */
@Slf4j
@Service
public class SearchService implements ISearchService {
    @Autowired
    private HouseRepo houseRepo;
    @Autowired
    private HouseDetailRepo detailRepo;
    @Autowired
    private HouseTagRepo tagRepo;
    @Autowired
    private ModelMapper modelMapper;
    @Autowired
    private TransportClient esClient;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private SupportAddressRepository addressRepository;
    @Autowired
    private IAddressService addressService;

    private static String INDEX_NAME="xunwu";
    private static String INDEX_TYPE="house";
    public static final String INDEX_TOPIC="house_build";
    public static final String ALL="*";

    @KafkaListener(topics = INDEX_TOPIC)
    private void handleMessage(String content){
        try {
            HouseIndexMessage message = objectMapper.readValue(content, HouseIndexMessage.class);
            switch (message.getOperation()){
                case HouseIndexMessage.INDEX:
                    createOrUpdateIndex(message);
                    break;
                case HouseIndexMessage.REMOVE:
                    removeIndex(message);
                    break;
                default:
                    log.warn("not support method"+message);
            }
        } catch (IOException e) {
            log.error("can not parse json ");
        }

    }

    private void removeIndex(HouseIndexMessage message) {
        DeleteByQueryRequestBuilder delete = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, message.getHouseId()))
                .source(INDEX_NAME);
        if(delete.get().getDeleted()<=0){
            this.remove(message.getHouseId(),message.getRetry()+1);
        }

    }

    private void remove(Integer houseId, int retry) {
        if(retry>HouseIndexMessage.MAX_RETRY){
            log.error("删除索引错误"+houseId);
            return;
        }
        HouseIndexMessage message = new HouseIndexMessage(houseId,HouseIndexMessage.REMOVE,retry);
        try {
            kafkaTemplate.send(INDEX_TOPIC,objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            log.error("parse error");
        }
    }


    private void createOrUpdateIndex(HouseIndexMessage message){
        Integer houseId = message.getHouseId();
        List<String> tagNames = new ArrayList<>();
        House house = houseRepo.findOne(houseId);
        if(house==null){
            log.error("房屋建索引失败,houseId="+houseId);
            this.index(houseId,message.getRetry()+1);
            return;
        }
        HouseIndexTemplate template =  new HouseIndexTemplate();
        modelMapper.map(house,template);
        HouseDetail houseDetail = detailRepo.findByHouseId(houseId);
        if(houseDetail==null){
            log.warn("房屋详情查询失败 :"+houseId);
        }
        modelMapper.map(houseDetail,template);

        SupportAddress city = addressRepository.findByEnNameAndLevel(house.getCityEnName(), SupportAddress.Level.CITY.getValue());
        SupportAddress region = addressRepository.findByEnNameAndLevel(house.getRegionEnName(), SupportAddress.Level.REGION.getValue());

        String address = city.getCnName()+region.getCnName()+house.getStreet()+house.getDistrict()+houseDetail.getAddress();

        HouseResult<BaiduMapLocation> baiduMapLocation = addressService.getBaiduMapLocation(city.getCnName(), address);

        if(!baiduMapLocation.isSuccess()){
            this.index(message.getHouseId(),message.getRetry()+1);
            return;
        }
        template.setLocation(baiduMapLocation.getResult());

        List<HouseTag> tags = tagRepo.findAllByHouseId(houseId);
        if(tags.isEmpty()){
            log.warn("房屋标签为空 :" +houseId);
        }
        tags.forEach(i->tagNames.add(i.getName()));
        template.setTags(tagNames);
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                .setQuery(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId));
        log.debug(requestBuilder.toString());
        SearchResponse searchResponse = requestBuilder.get();

        boolean success=false;
        long totalHits = searchResponse.getHits().getTotalHits();
        if(totalHits==0){
            try {
                success=create(template);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }else if(totalHits==1){
            String id = searchResponse.getHits().getAt(0).getId();
            try {
                success=update(id,template);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }else{
            try {
                success=deleteAndUpdate(totalHits,template);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }

        if(success){
            log.debug("index success "+houseId);
        }

    }

    private void index(Integer houseId, int retry) {
        if(retry>HouseIndexMessage.MAX_RETRY){
            log.error("重试三次失败。丢弃消息"+houseId);
            return;
        }
        HouseIndexMessage message = new HouseIndexMessage(houseId,HouseIndexMessage.INDEX,retry);
        try {
            kafkaTemplate.send(INDEX_TOPIC,objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            log.error("parse error");
        }
    }


    /**
     * 从mysql中查出数据 然后去es构建索引
     * @param houseId
     */
    @Override
    public void index(Integer houseId) {
        index(houseId,0);
    }

    @Override
    public void remove(Integer houseId) {
        remove(houseId,0);
    }




    @Override
    public ServiceResult query(RentSearchForm searchForm) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,searchForm.getCityEnName()));
        if(searchForm.getRegionEnName()!=null&&!ALL.equals(searchForm.getRegionEnName())){
            boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,searchForm.getRegionEnName()));
        }
        //关键词匹配


        if(searchForm.getKeywords()!=null&&!searchForm.getKeywords().isEmpty()) {

            boolQuery.must(
                    QueryBuilders.matchQuery(HouseIndexKey.TITLE,searchForm.getKeywords()).boost(2.0f)
            );

            boolQuery.should(QueryBuilders.multiMatchQuery(searchForm.getKeywords(),
                    HouseIndexKey.TRAFFIC,
                    HouseIndexKey.DISTRICT,
                    HouseIndexKey.ROUND_SERVICE,
                    HouseIndexKey.SUBWAY_LINE_NAME,
                    HouseIndexKey.SUBWAY_STATION_NAME
            ));
        }
        //条件筛选
        RentValueBlock matchArea = RentValueBlock.matchArea(searchForm.getAreaBlock());
        RentValueBlock matchPrice = RentValueBlock.matchPrice(searchForm.getPriceBlock());
        //面积
        if(!RentValueBlock.ALL.equals(matchArea)){
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery(HouseIndexKey.AREA);
            if(matchArea.getMax()>0){
                rangeQuery.lte(matchArea.getMax());
            }
            if(matchArea.getMin()>0){
                rangeQuery.gte(matchArea.getMin());
            }
            boolQuery.filter(rangeQuery);
        }
        //价格
        if(!RentValueBlock.ALL.equals(matchPrice)){
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery(HouseIndexKey.PRICE);
            if(matchPrice.getMax()>0){
                rangeQuery.lte(matchPrice.getMax());
            }
            if(matchPrice.getMin()>0){
                rangeQuery.gte(matchPrice.getMin());
            }
            boolQuery.filter(rangeQuery);
        }
        //朝向
        if(searchForm.getDirection()>0){
            boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.DIRECTION,searchForm.getDirection()));
        }
        //租用方式
        if(searchForm.getRentWay()>-1)
        {
            boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.RENT_WAY,searchForm.getRentWay()));
        }
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(
                        HouseSort.getSortKey(searchForm.getOrderBy()), SortOrder.fromString(searchForm.getOrderDirection())
                )
                .setFrom(searchForm.getStart())
                .setSize(searchForm.getSize())
                .setFetchSource(HouseIndexKey.HOUSE_ID,null);
        System.out.println(requestBuilder.toString());
        SearchResponse response = requestBuilder.get();
        List<Integer> ids = new ArrayList<>();

        if(response.status()!=RestStatus.OK){
            return new ServiceResult(ids);
        }

        response.getHits().forEach(i->ids.add(Ints.tryParse(i.getSource().get(HouseIndexKey.HOUSE_ID).toString())));

        return new ServiceResult(response.getHits().getTotalHits(),ids);
    }

    @Override
    public HouseResult autoComplete(String prefix) {
        CompletionSuggestionBuilder builder = SuggestBuilders.completionSuggestion("suggest").prefix(prefix).size(5);
        SuggestBuilder suggestBuilder = new SuggestBuilder();
        suggestBuilder.addSuggestion("autocomplete",builder);
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .suggest(suggestBuilder);
        SearchResponse response = requestBuilder.get();
        Suggest suggest = response.getSuggest();
        Suggest.Suggestion result = suggest.getSuggestion("autocomplete");
        int max = 5;
        Set<String> sugSet = new HashSet<>();
        for (Object res : result) {
            if(res instanceof CompletionSuggestion.Entry){
                CompletionSuggestion.Entry item = (CompletionSuggestion.Entry) res;
                if(item.getOptions().isEmpty()){
                    continue;
                }
                for (CompletionSuggestion.Entry.Option option : item.getOptions()) {
                    String tips = option.getText().toString();
                    if(sugSet.contains(tips)){
                        continue;
                    }
                    sugSet.add(tips);
                    max++;
                }
            }
           if(max>5){
                break;
           }
        }

        return HouseResult.of(Lists.newArrayList(sugSet.toArray(new String[]{})));
    }

    @Override
    public HouseResult aggregateDistrictHouse(String cityEnName, String regionEnName, String district) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders
                        .boolQuery()
                        .filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName))
                        .filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,regionEnName))
                        .filter(QueryBuilders.termQuery(HouseIndexKey.DISTRICT,district));
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQueryBuilder)
                .addAggregation(
                        AggregationBuilders.terms(HouseIndexKey.AGG_DISTRICT)
                                .field(HouseIndexKey.DISTRICT))
                .setSize(0);
        SearchResponse response = requestBuilder.get();
        if(response.status()==RestStatus.OK){
            Terms terms  = response.getAggregations().get(HouseIndexKey.AGG_DISTRICT);
            if(terms.getBuckets()!=null&&!terms.getBuckets().isEmpty()){
                return HouseResult.of(terms.getBucketByKey(district).getDocCount());
            }
        }else{
            log.warn("数据聚合失败");

        }
        return HouseResult.of(0);
    }

    @Override
    public ServiceResult aggCityHouse(String cityEnName) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));
        TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms(HouseIndexKey.AGG_REGION).field(HouseIndexKey.REGION_EN_NAME);
        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE).setQuery(boolQueryBuilder).addAggregation(aggregationBuilder);
        log.info(requestBuilder.toString());
        SearchResponse response = requestBuilder.get();
        List<HouseBucketDTO> bucketDTOS = new ArrayList<>();
        if(response.status()!=RestStatus.OK){
            log.warn("聚合失败"+requestBuilder);
            return new ServiceResult(bucketDTOS);
        }
        Terms terms = response.getAggregations().get(HouseIndexKey.AGG_REGION);
        terms.getBuckets().forEach(i->{
            bucketDTOS.add(new HouseBucketDTO(i.getKeyAsString(),i.getDocCount()));
        });
        return new ServiceResult(bucketDTOS.get(0).getCount(),bucketDTOS);
    }

    @Override
    public ServiceResult mapQuery(String cityEnName, String orderBy, String orderDirection, int start, int size) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));
        SearchRequestBuilder searchRequestBuilder = this.esClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE).setQuery(boolQueryBuilder)
                .addSort(HouseSort.getSortKey(orderBy), SortOrder.fromString(orderDirection))
                .setFrom(start)
                .setSize(size);
        List ids  = new ArrayList();
        SearchResponse response = searchRequestBuilder.get();
        if(response.status()!=RestStatus.OK){
            log.warn("search is not ok");
            return new ServiceResult(ids);
        }
        response.getHits().forEach(i->ids.add(Ints.tryParse((i.getSource().get(HouseIndexKey.HOUSE_ID)).toString())));
        return new ServiceResult(response.getHits().getTotalHits(),ids);
    }

    @Override
    public ServiceResult mapQuery(MapSearch mapSearch) {
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,mapSearch.getCityEnName()));
        queryBuilder.filter(
                QueryBuilders.geoBoundingBoxQuery("location").
                        setCorners(
                                new GeoPoint(mapSearch.getLeftLatitude(),mapSearch.getLeftLongitude()),
                        new GeoPoint(mapSearch.getRightLatitude(),mapSearch.getRightLongitude()))
        );
        SearchRequestBuilder searchRequestBuilder = this.esClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE).setQuery(queryBuilder).addSort(HouseSort.getSortKey(mapSearch.getOrderBy()),
                SortOrder.fromString(mapSearch.getOrderDirection()))
                .setFrom(mapSearch.getStart())
                .setSize(mapSearch.getSize());
        List ids  = new ArrayList();
        SearchResponse response = searchRequestBuilder.get();
        if(response.status()!=RestStatus.OK){
            log.warn("search is not ok");
            return new ServiceResult(ids);
        }
        response.getHits().forEach(i->ids.add(Ints.tryParse((i.getSource().get(HouseIndexKey.HOUSE_ID)).toString())));
        return new ServiceResult(response.getHits().getTotalHits(),ids);
    }

    private boolean updateSuggest(HouseIndexTemplate template){
        AnalyzeRequestBuilder requestBuilder = new AnalyzeRequestBuilder(this.esClient, AnalyzeAction.INSTANCE,INDEX_NAME,template.getTitle(),template.getLayoutDesc(),template.getRoundService(),template.getDescription(),template.getSubwayStationName(),template.getSubwayLineName());
        requestBuilder.setAnalyzer("ik_smart");
        AnalyzeResponse response = requestBuilder.get();
        List<AnalyzeResponse.AnalyzeToken> tokens = response.getTokens();
        if(tokens==null){
            log.warn("分词错误");
            return false;
        }
        List<HouseSuggest> suggests = new ArrayList<>();
        for (AnalyzeResponse.AnalyzeToken token : tokens) {
            //排除数字类型以及长度小于2的结果
            if("<NUM>".equals(token.getType())||token.getTerm().length()<2){
                continue;
            }
            HouseSuggest suggest = new HouseSuggest();
            suggest.setInput(token.getTerm());
            suggests.add(suggest);
        }
        template.setSuggest(suggests);
        return true;
    }


    private boolean create(HouseIndexTemplate template) throws JsonProcessingException {
        if(!updateSuggest(template)){
            return false;
        }
        IndexResponse response = this.esClient.prepareIndex(INDEX_NAME, INDEX_TYPE)
                .setSource(objectMapper.writeValueAsBytes(template), XContentType.JSON).get();
        log.debug("create Index :"+template.getHouseId());
        if(response.status()== RestStatus.CREATED){
            return true;
        }else{
            return false;
        }
    }
    private boolean update(String esId,HouseIndexTemplate template) throws JsonProcessingException {
        if(!updateSuggest(template)){
            return false;
        }
        UpdateResponse response = this.esClient.prepareUpdate(INDEX_NAME, INDEX_TYPE, esId)
                .setDoc(objectMapper.writeValueAsBytes(template), XContentType.JSON).get();
        log.debug("update Index :"+template.getHouseId());
        if(response.status()== RestStatus.CREATED){
            return true;
        }else{
            return false;
        }
    }

    private boolean deleteAndUpdate(Long totalHit,HouseIndexTemplate template) throws JsonProcessingException {
        DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, template.getHouseId()))
                .source(INDEX_NAME);
        log.debug("delete by query "+builder.get());
        BulkByScrollResponse response = builder.get();
        long deleted = response.getDeleted();
        if(deleted!=totalHit){
            log.warn("删除数目不一致:{}-{}",totalHit,deleted);
            return false;
        }
        return create(template);
    }


}
